package com.stephen.redisdemo.atomic;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author: stephen
 * @create: 2020-03-01 12:12
 * @description: 用阻塞队列,实现生产者消费者模式
 */
class ShareData {
    private volatile boolean FLAG = true;
    // 操作的资源
    private AtomicInteger atomicInteger = new AtomicInteger();
    private BlockingQueue<String> blockingQueue = null;

    public ShareData(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    public void product() throws InterruptedException {
        String data = null;
        while (FLAG) {
            data = atomicInteger.incrementAndGet() + "";
            blockingQueue.offer(data);
            System.out.println(Thread.currentThread().getName() + "\t" + "开始生产" + data);
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName() + "\t" + "停止生产");
    }

    public void consumer() throws InterruptedException {
        String data = null;
        while (FLAG) {
            data = blockingQueue.poll(2,TimeUnit.SECONDS);
            if (data == null) {
                FLAG = false;
                System.out.println(Thread.currentThread().getName() + "\t" + "停止消费");
                return;
            }
            System.out.println(Thread.currentThread().getName() + "\t" + "消费" + data);
        }
    }

    public void stop() {
        FLAG = false;
    }

}

public class Pro_Con_BlockDemo {

    public static void main(String[] args) throws InterruptedException {
        ShareData shareData = new ShareData(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            try {
                shareData.product();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"prod").start();
        new Thread(() -> {
            try {
                shareData.consumer();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"consumer").start();
        TimeUnit.SECONDS.sleep(5);
        shareData.stop();
    }
}
